package com.briup.prs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * inputformat KeyValueTextInoutFormat
 * map 阶段原样输出  Mapper
 * reduce  自定义  Step6Reducer
 * outputformat   默认  TextOutPutFormat
 */
public class Step6 extends Configured implements Tool {
    public static class Step6Reducer extends Reducer<Text, Text, Text, DoubleWritable> {

        /**
         * @param key     shopA
         * @param values  [{user:v,……},{shopB-count,……}]
         * @param context
         * @throws IOException
         * @throws InterruptedException 输出 user:shopB  v*count
         */
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//            保存共现矩阵的数据
            Map<String, Integer> m1 = new HashMap<>();
//          保存偏好矩阵的数据
            Map<String, Double> m2 = new HashMap<>();
//            遍历矩阵  只有两个  一个是共现  一个是偏好
            for (Text value : values) {
//                将矩阵字符串中的内容（每一个元素） 都切出来
                String str = value.toString();
                String[] arr = str.split("[,]");
//                判断本次循环读取到的是哪个矩阵 如果是 "-" 共现矩阵  如果是":" 偏好矩阵
                if (str.contains("-")) {
                    for (String s : arr) {
                        String[] data = s.split("-");
//                        将共现商品放到 m1的key  共现次数放到m1的value位置
                        m1.put(data[0], Integer.parseInt(data[1]));
                    }
                } else {
                    for (String s : arr) {
                        String[] data = s.split(":");
//                        将用户放到 m2的key  偏好值放到m2的value位置
                        m2.put(data[0], Double.parseDouble(data[1]));
                    }

                }
            }

//            遍历两个map集合  要其key拼接  value相乘
            m1.forEach((km1,vm1)->{
                m2.forEach((km2,vm2)->{
//                    拼接用户id和商品
                   String outKey= km2+":"+km1;
//                   计算推荐值
                   double outVal=vm1*vm2;
                    try {
                        context.write(new Text(outKey),new DoubleWritable(outVal));
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });

            });
        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String in1 = conf.get("in1");
        String in2 = conf.get("in2");
        String out = conf.get("out");

        Job job = Job.getInstance(conf);
        job.setJobName("Step6");
        job.setJarByClass(this.getClass());

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.setInputPaths(job,new Path(in1),new Path(in2));

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(Step6Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        TextOutputFormat.setOutputPath(job,new Path(out));
        return job.waitForCompletion(true)?0:-1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Step6(), args);
    }
}
